package com.ml.data;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;

import com.alibaba.fastjson.JSON;
import com.ml.enums.EnumSymbol;
import com.ml.message.OrderDTO;
import com.ml.message.TradeDTO;
import com.ml.plugins.rocketmq.MatchSink;
import com.ml.util.MathUtils;
import com.virgil.redis.util.RedisUtil;

import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;

@EnableBinding(MatchSink.class)
@Slf4j
public class MatchOutResources {

	@Autowired
	RedisUtil redisUtil;

	private String depthStr = "记录上次深度";

//    /**
//     * @param echo
//     * @return void 返回类型
//     * @throws
//     * @Title: push_depth
//     * @Description: TODO(盘口深度数据)
//     */
//    @StreamListener(MatchSink.IN_PUSH_DEPTH)
//    public void push_depth(Flux<String> flux) {
//        flux.subscribe(message -> {
//            if (!depthStr.equals(message)) {
//                depthStr = message;
//                JSONObject data = JSONObject.parseObject(message);
//                String coinTeam = data.getString("coinTeam");
//                redisUtil.set(StringUtils.join(coinTeam, RedisKey.DEPTH_SUFFIX), message);
//            }
//        });
//    }

	/**
	 * @param echo
	 * @return void 返回类型
	 * @throws @Title: push_depth
	 * @Description: TODO(订单变化)
	 */
	@StreamListener(MatchSink.IN_ORDER_ALTER)
	public void update_order(OrderDTO order) {
		log.info("订单变化" + order);
	}

	/**
	 * @return void 返回类型
	 * @throws @Title: push_depth
	 * @Description: TODO(新的交易记录)
	 */
	@StreamListener(MatchSink.IN_NEW_TRADE)
	public void new_trade(TradeDTO trade) {
//        flux.subscribe(message -> {
		log.info("交易记录：" + trade);
		// 获取交易对
		EnumSymbol enumSymbol = EnumSymbol.of(trade.getSymbolId()).get();
		String symbol = enumSymbol.getOne() + "_" + enumSymbol.getTwo();
		// 更新最新价格
		redisUtil.set(symbol.concat(":ExchangePrice"), MathUtils.LongToBigDecimal(trade.getPrice(), 9).toPlainString());
		// 缓存成交记录
		redisUtil.lpush(0, symbol.concat(":exchange"), JSON.toJSONString(trade));
		redisUtil.ltrim(symbol.concat(":exchange"), 0, 50);
		// 刷新K线数据

//        });
	}
}
